package com.apache.flink.basic

import com.apache.flink.bean.Domain.Access
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.Table
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.types.Row

// 开发时要做的第一件事情
import org.apache.flink.api.scala._
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.table.api._


object DataStreamTableSQLApp02 {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val tableEnvironment = StreamTableEnvironment.create(env)

    val stream = env.readTextFile("data/access.log")
      .map(x => {
        val splits = x.split(",")
        Access(splits(0).trim.toLong, splits(1).trim, splits(2).trim.toDouble)
      })

    // datastream ==> table
    val accessTable: Table = tableEnvironment.fromDataStream(stream)
    tableEnvironment.createTemporaryView("access", accessTable)

    val resultTable: Table = tableEnvironment
      .sqlQuery("select domain, sum(traffic) as traffics from access group by domain")

    /**
     * Exception in thread "main" org.apache.flink.table.api.TableException:
     * toAppendStream doesn't support consuming update changes
     * which is produced by node GroupAggregate(groupBy=[domain],
     * select=[domain, SUM(traffic) AS traffics])
     */
    tableEnvironment.toRetractStream[Row](resultTable).print()

//    accessTable.groupBy("domain")
//        .aggregate("sum(traffic) as traffics")
//        .select("domain, traffics")
//      .toRetractStream[Row].filter(_._1).print("过时写法...")

//    accessTable.groupBy('domain)
//        .aggregate('traffic.sum().as("traffics"))
//        .select('domain, 'traffics)
//      .toRetractStream[Row].filter(_._1).print("推荐写法...")


    env.execute(getClass.getCanonicalName)
  }
}
